package com.apex.flink.sink;

import com.apex.flink.FlinkEnvironment;
import com.apex.flink.utils.ConfigKeyName;
import com.apex.flink.stream.FlinkStreamSink;
import com.typesafe.config.Config;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.types.Row;

public class RedisStreamSink implements FlinkStreamSink<Row,Row> {
    private Config config;
    private String hosts;
    private String clusterType;
    @Override
    public void prepare(FlinkEnvironment plugin) {
        if (config.hasPath(ConfigKeyName.SINK_REDIS_HOSTS)){
            hosts = config.getString(ConfigKeyName.SINK_REDIS_HOSTS);
            clusterType = config.getString(ConfigKeyName.SINK_REDIS_TYPE);
        }
    }

    @Override
    public Config getConfig() {
        return config;
    }

    @Override
    public void setConfig(Config config) {
        this.config = config;
    }

    @Override
    public DataStreamSink<Row> outputStream(FlinkEnvironment env, DataStream<Row> dataStream) {
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost(hosts).build();
        DataStream<Tuple2<String, String>> stream = dataStream.map(new MapFunction<Row, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(Row value) throws Exception {
                String key = String.valueOf(value.getField(0));
                return new Tuple2<String, String>(key,value.toString());
            }
        });
        stream.addSink(new RedisSink<Tuple2<String, String>>(conf,new RedisSingleMapper()));
        return dataStream.addSink(null);
    }
    //
    public static class RedisSingleMapper implements RedisMapper<Tuple2<String, String>> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "HASH_NAME");
        }

        @Override
        public String getKeyFromData(Tuple2<String, String> data) {
            return data.f0;
        }

        @Override
        public String getValueFromData(Tuple2<String, String> data) {
            return data.f1;
        }
    }
}

